package com.zcy;

import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.runtime.client.EPEventService;
import com.zcy.event.MyEvent;
import com.zcy.util.EsperBuilder;
import com.zcy.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 14.3.5. Time batch Window (time_batch or win:time_batch)
 * 称为滚动窗口（Tumbling Window)
 * 滚动窗口也为[左闭，右闭]的方式，当一个窗口触发完毕之后，则窗口中的数据进行清空
 * <p>
 * TODO 对于时间类别滚动窗口有如下法则：
 * TODO 如果未指定时间基准，则将第一个事件的事件为基准； 这一条法则只针对Batch类型的滚动窗口有用，对于滑动窗口效果是一样的
 * TODO 因此在所有Time Batch，都有一个参数：flow_control(流控制)，而length Batch与其他非Batch(滑动窗口)中均没有flow_control参数
 */
@Slf4j
public class TimeBatchWindow extends TimeUtils {

    private static final String _TimeBatch = "@name('time-statement') select sum(num) from MyEvent#win:time_batch(2 sec)";
    private static final String _TimeDelayBatch = "@name('time-statement') select sum(num) from MyEvent#win:time_batch(2 sec, 3000L)";


    /**
     * 与Time滑动窗口比较最大的区别在于，进入滚动窗口的数据不会提前触发，直到窗口时间符合被整个触发，
     * 当窗口触发之后，会清空当前窗口数据
     * sum is : 30
     * sum is : null
     * sum is : 30
     * sum is : null
     * sum is : 40
     * sum is : null
     * <p>
     * TODO：如果未指定时间基准，则将第一个事件的事件为基准； 这一条法则只针对Batch类型的滚动窗口有用，对于滑动窗口效果是一样的
     */
    @Test
    public void lengthWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_TimeBatch, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        });

        sleep(TimeUnit.SECONDS, 5L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 1899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 2899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 3899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }


    /**
     * 指定时间对齐，而没有使用默认的第一个时间的时间为基准时间
     * Time_Batch的第二个参数有点FLink窗口对齐的意思
     * FLink窗口对齐公式：WindowStartWithOffset = timestamp - (timestamp - offset + windowSize) % windowSize
     * sum is : 10
     * sum is : 20
     * sum is : null
     * sum is : 30
     * sum is : null
     * sum is : 40
     * sum is : null
     */
    @Test
    public void delayTimeBatch() {
        EsperBuilder esperBuilder = new EsperBuilder(_TimeDelayBatch, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> Stream.of(eventBeans).forEach(item -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        }));

        sleep(TimeUnit.SECONDS, 5L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 1899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 2899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.MILLISECONDS, 3899L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }


}
